package com.dy.dws.TradeHospPatGenAGConsulPaySucWin;

import cn.hutool.core.collection.ListUtil;
import com.alibaba.fastjson.JSONObject;
import com.dy.dws.function.MyAsyncDataStream;
import com.dy.realtime.common.base.BaseApp;
import com.dy.realtime.common.bean.DwsTradeHospPatGenAGConsulPaySucBean;
import com.dy.realtime.common.constant.Constant;
import com.dy.realtime.common.util.DateFormatUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.concurrent.TimeUnit;

/**
 * @projectName: MedicalTreatment
 * @package: com.dy.dws.TradeHospPatGenAGConsulPaySucWin
 * @className: DwsTradeHospPatGenAGConsulPaySucWin
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/14 10:21
 * @version: 1.0
 * 交易域医院性别年龄段粒度问诊支付成功各窗口汇总表
 */

public class DwsTradeHospPatGenAGConsulPaySucWin extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeHospPatGenAGConsulPaySucWin().start(Constant.TOPIC_DWD_TRADE_CONSULTATION_PAY_SUC, Constant.TOPIC_DWS_TRADE_HOSP_PAT_GEN_A_G_CONSUL_WIN, 4, 10020);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        dataStreamSource.print();
        //对数据进行elt处理 过滤出 doctorId 和 patient不为null得数据
        SingleOutputStreamOperator<JSONObject> etlStream = getEtlStream(dataStreamSource);
//        etlStream.print();
        //转换为实体类
        SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulPaySucBean> beanStream = getBeanStream(etlStream);
//        beanStream.print();
        //异步io连接 连接医生表 获取医院id
        SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulPaySucBean> getHospitalId = getHospitalId(beanStream);
        //连接医院表获取医院名称
        SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulPaySucBean> getHospitalName = getHospitalName(getHospitalId);
//        getHospitalName.print();
        //连接就诊人表 获取性别编码 以及 计算年龄段
        SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulPaySucBean> getPatient = getPatient(getHospitalName);
//        getPatient.print();
        //连接字典表 获取性别名称
//        getGenderValue.print();
        //按照医院 性别 年龄段进行keyby 并设置水位线
        KeyedStream<DwsTradeHospPatGenAGConsulPaySucBean, Tuple3<String, String, String>> wmStream = getWmStream(getPatient);
//        wmStream.print();
        //开窗聚合
        SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulPaySucBean> result = getProcess(wmStream);
        SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulPaySucBean> getGenderValue = getGenderValue(result);
        getGenderValue.print();
        //写入doris
//        getGenderValue.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink("dws_trade_hosp_pat_gen_a_g_consul_pay_suc_win"));
    }

    private static SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulPaySucBean> getProcess(KeyedStream<DwsTradeHospPatGenAGConsulPaySucBean, Tuple3<String, String, String>> wmStream) {
        return wmStream.window(TumblingEventTimeWindows.of(Time.seconds(5))).process(new ProcessWindowFunction<DwsTradeHospPatGenAGConsulPaySucBean, DwsTradeHospPatGenAGConsulPaySucBean, Tuple3<String, String, String>, TimeWindow>() {
            @Override
            public void process(Tuple3<String, String, String> stringStringStringTuple3, ProcessWindowFunction<DwsTradeHospPatGenAGConsulPaySucBean, DwsTradeHospPatGenAGConsulPaySucBean, Tuple3<String, String, String>, TimeWindow>.Context context, Iterable<DwsTradeHospPatGenAGConsulPaySucBean> iterable, Collector<DwsTradeHospPatGenAGConsulPaySucBean> collector) throws Exception {
                ArrayList<DwsTradeHospPatGenAGConsulPaySucBean> sucBeans = ListUtil.toList(iterable);
                double sum = sucBeans.stream().mapToDouble(e -> e.getConsultationPaySucAmount().doubleValue()).sum();
                long SucCount = sucBeans.stream().mapToLong(e -> e.getConsultationPaySucCount()).sum();
                String start = DateFormatUtil.tsToDateTime(context.window().getStart());
                String end = DateFormatUtil.tsToDateTime(context.window().getEnd());
                String curDate = DateFormatUtil.tsToDate(System.currentTimeMillis());
                DwsTradeHospPatGenAGConsulPaySucBean next = sucBeans.iterator().next();
                DwsTradeHospPatGenAGConsulPaySucBean result = DwsTradeHospPatGenAGConsulPaySucBean.builder()
                        .hospitalId(next.getHospitalId())
                        .ageGroup(next.getAgeGroup())
                        .consultationPaySucAmount(new BigDecimal(sum))
                        .consultationPaySucCount(SucCount)
                        .stt(start)
                        .edt(end)
                        .curDate(curDate)
                        .userId(next.getId())
                        .consultationPaySucTime(next.getConsultationPaySucTime())
                        .consultationFee(next.getConsultationFee())
                        .doctorId(next.getDoctorId())
                        .hospitalName(next.getHospitalName())
                        .patientId(next.getPatientId())
                        .patientGenderCode(next.getPatientGenderCode())
                        .patientGender(next.getPatientGender())
                        .id(next.getId())
                        .build();
                collector.collect(result);

            }
        });
    }

    private static KeyedStream<DwsTradeHospPatGenAGConsulPaySucBean, Tuple3<String, String, String>> getWmStream(SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulPaySucBean> getGenderValue) {
        return getGenderValue.assignTimestampsAndWatermarks(WatermarkStrategy.<DwsTradeHospPatGenAGConsulPaySucBean>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner(new SerializableTimestampAssigner<DwsTradeHospPatGenAGConsulPaySucBean>() {
            @Override
            public long extractTimestamp(DwsTradeHospPatGenAGConsulPaySucBean dwsTradeHospPatGenAGConsulPaySucBean, long l) {
                return dwsTradeHospPatGenAGConsulPaySucBean.getTs();
            }
        })).keyBy(new KeySelector<DwsTradeHospPatGenAGConsulPaySucBean, Tuple3<String, String, String>>() {
            @Override
            public Tuple3<String, String, String> getKey(DwsTradeHospPatGenAGConsulPaySucBean dwsTradeHospPatGenAGConsulPaySucBean) throws Exception {
                return Tuple3.of(dwsTradeHospPatGenAGConsulPaySucBean.getHospitalName(), dwsTradeHospPatGenAGConsulPaySucBean.getPatientGender(), dwsTradeHospPatGenAGConsulPaySucBean.getAgeGroup());
            }
        });
    }


    private static SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulPaySucBean> getGenderValue(SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulPaySucBean> getPatient) {
        return AsyncDataStream.unorderedWait(getPatient, new MyAsyncDataStream<DwsTradeHospPatGenAGConsulPaySucBean>("dim_dict") {
            @Override
            public void getResult(DwsTradeHospPatGenAGConsulPaySucBean dwsTradeHospPatGenAGConsulPaySucBean, JSONObject jsonObject) {
                dwsTradeHospPatGenAGConsulPaySucBean.setPatientGender(jsonObject.getString("value"));
            }

            @Override
            public String getDimId(DwsTradeHospPatGenAGConsulPaySucBean dwsTradeHospPatGenAGConsulPaySucBean) {
                return dwsTradeHospPatGenAGConsulPaySucBean.getPatientGenderCode();
            }
        }, 100, TimeUnit.MINUTES, 100);
    }


    private static SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulPaySucBean> getPatient(SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulPaySucBean> getHospitalName) {
        return AsyncDataStream.unorderedWait(getHospitalName, new MyAsyncDataStream<DwsTradeHospPatGenAGConsulPaySucBean>("dim_patient") {
            @Override
            public void getResult(DwsTradeHospPatGenAGConsulPaySucBean dwsTradeHospPatGenAGConsulPaySucBean, JSONObject jsonObject) {
                String birthday = jsonObject.getString("birthday");
                DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
                String s = DateFormatUtil.tsToDate(System.currentTimeMillis());
                LocalDate parse = LocalDate.parse(s, formatter);
                int year1 = parse.getYear();
                int age;
                String ageGroup = null;
                if (birthday != null) {
                    LocalDate birthDate = LocalDate.parse(birthday, formatter);
                    int year = birthDate.getYear();
                    age = year1 - year;
                    if (age >= 0 && age <= 2) {
                        ageGroup = "婴儿期";
                    } else if (age <= 5) {
                        ageGroup = "幼儿期";
                    } else if (age <= 11) {
                        ageGroup = "小学阶段";
                    } else if (age <= 17) {
                        ageGroup = "青少年期";
                    } else if (age <= 29) {
                        ageGroup = "青年期";
                    } else if (age <= 59) {
                        ageGroup = "中年期";
                    } else if (age > 60) {
                        ageGroup = "老年期";
                    } else {
                        ageGroup = "年龄异常";
                    }
                }
                dwsTradeHospPatGenAGConsulPaySucBean.setAgeGroup(ageGroup);
                dwsTradeHospPatGenAGConsulPaySucBean.setPatientGenderCode(jsonObject.getString("gender"));

            }

            @Override
            public String getDimId(DwsTradeHospPatGenAGConsulPaySucBean dwsTradeHospPatGenAGConsulPaySucBean) {
                return dwsTradeHospPatGenAGConsulPaySucBean.getPatientId();
            }
        }, 100, TimeUnit.MINUTES, 100);
    }

    private static SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulPaySucBean> getHospitalName(SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulPaySucBean> getHospitalId) {
        return AsyncDataStream.unorderedWait(getHospitalId, new MyAsyncDataStream<DwsTradeHospPatGenAGConsulPaySucBean>("dim_hospital") {
            @Override
            public void getResult(DwsTradeHospPatGenAGConsulPaySucBean dwsTradeHospPatGenAGConsulPaySucBean, JSONObject jsonObject) {
                dwsTradeHospPatGenAGConsulPaySucBean.setHospitalName(jsonObject.getString("name"));
            }

            @Override
            public String getDimId(DwsTradeHospPatGenAGConsulPaySucBean dwsTradeHospPatGenAGConsulPaySucBean) {
                return dwsTradeHospPatGenAGConsulPaySucBean.getHospitalId();
            }
        }, 100, TimeUnit.MINUTES, 100);
    }

    private static SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulPaySucBean> getHospitalId(SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulPaySucBean> beanStream) {
        return AsyncDataStream.unorderedWait(beanStream, new MyAsyncDataStream<DwsTradeHospPatGenAGConsulPaySucBean>("dim_doctor") {
            @Override
            public void getResult(DwsTradeHospPatGenAGConsulPaySucBean dwsTradeHospPatGenAGConsulPaySucBean, JSONObject jsonObject) {
                dwsTradeHospPatGenAGConsulPaySucBean.setHospitalId(jsonObject.getString("hospital_id"));
            }

            @Override
            public String getDimId(DwsTradeHospPatGenAGConsulPaySucBean dwsTradeHospPatGenAGConsulPaySucBean) {
                return dwsTradeHospPatGenAGConsulPaySucBean.getDoctorId();
            }
        }, 100, TimeUnit.MINUTES, 100);
    }

    private static SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulPaySucBean> getBeanStream(SingleOutputStreamOperator<JSONObject> etlStream) {
        return etlStream.map(new MapFunction<JSONObject, DwsTradeHospPatGenAGConsulPaySucBean>() {
            @Override
            public DwsTradeHospPatGenAGConsulPaySucBean map(JSONObject jsonObject) throws Exception {
                DwsTradeHospPatGenAGConsulPaySucBean dwsTradeHospPatGenAGConsulPaySucBean = JSONObject.parseObject(jsonObject.toString(), DwsTradeHospPatGenAGConsulPaySucBean.class);
                dwsTradeHospPatGenAGConsulPaySucBean.setConsultationPaySucCount(1L);
                dwsTradeHospPatGenAGConsulPaySucBean.setConsultationPaySucAmount(dwsTradeHospPatGenAGConsulPaySucBean.getConsultationFee());
                return dwsTradeHospPatGenAGConsulPaySucBean;
            }
        });
    }

    private static SingleOutputStreamOperator<JSONObject> getEtlStream(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.flatMap(new RichFlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                try {
                    JSONObject jsonObject = JSONObject.parseObject(s);                    String doctor_id = jsonObject.getString("doctor_id");
                    String patient_id = jsonObject.getString("patient_id");
                    if (doctor_id != null && patient_id != null) {
                        collector.collect(jsonObject);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }
}
